package com.benmei.ntlive.server.netty;
import com.benmei.ntlive.constant.Constants;
import com.benmei.ntlive.dto.TcpRequest;
import com.benmei.ntlive.dto.TcpRequestHead;
import com.benmei.ntlive.dto.TcpResponse;
import com.benmei.ntlive.dto.ntmsg.cons.Action;
import com.benmei.ntlive.entity.ClassRoom;
import com.benmei.ntlive.entity.ClassRoomEntryLog;
import com.benmei.ntlive.entity.Member;
import com.benmei.ntlive.entity.Teacher;
import com.benmei.ntlive.exception.NtClientException;
import com.benmei.ntlive.service.LiveService;
import com.benmei.ntlive.utils.ChannelUtil;
import com.benmei.ntlive.utils.JsonUtil;
import com.benmei.ntlive.vo.Room;
import com.benmei.ntlive.vo.Session;
import com.benmei.ntlive.vo.User;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.util.concurrent.ScheduledFuture;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * Create by chenhong on 2018/10/26 10:39
 * Description : 消息处理
 */
@Component
public class NetLiveHandler {
    private static Logger logger = LoggerFactory.getLogger(NetLiveHandler.class);

    @Autowired
    private LiveService liveService;

    @Autowired
    private SessionRepository sessionRepository;

    //业务处理
    public void handler(ChannelHandlerContext ctx, Object msg,int connectType){
        String stringMessage = (String) msg;
        try {
            // JSON格式校验
            if (!JsonUtil.isValidJSON(stringMessage)) {
                throw new NtClientException("Invalid JSON data format");
            }
            // 头部校验
            TcpRequest req = JsonUtil.toObject(stringMessage, TcpRequest.class);
            validateTcpHead(req.getHead());

            // action 校验
            Map<String, Object> body = req.getBody();
            Object objAction = body.get("action");
            if (objAction == null) {
                throw new NtClientException("Invalid data format:action is required");
            }
            if (!String.class.isInstance(objAction)) {
                throw new NtClientException("Invalid data format:action must be string");
            }
            String action = (String) objAction;
            if (!Action.isValidateAction(action)) {
                throw new NtClientException("Invalid action value '" + action + "'");
            }

            // 建立会话
            if (Action.CREATE_SESSION.getName().equalsIgnoreCase(action)) {
                // 校验RoomId
                Object objRoomId = req.getBody().get("roomId");
                if (objRoomId == null) {
                    throw new NtClientException("The parameter 'roomId' is required", action);
                }
                Integer roomId = getIntegerForRoomId(action, objRoomId);

                final ClassRoom classRoom = liveService.findRoomById(roomId);
                if (classRoom == null) {
                    throw new NtClientException("ClassRoom do not exists, The roomId=" + roomId, action);
                }
                createSession(ctx, req, action, roomId,connectType);

                /**
                 * resive by C.H on 2018/10/29 10:53
                 * 修改：
                 */
                ChannelUtil.get(connectType).writeAndFlush(ctx.channel(),JsonUtil.toJson(new TcpResponse(action)) + Constants.delimiter);
//                ctx.channel().writeAndFlush(JsonUtil.toJson(new TcpResponse(action)) + Constants.delimiter);

                // 当握手成功后，Login响应向下透传，主动发送心跳消息
                //NioEventLoop是一个Schedule,因此支持定时器的执行，创建心跳计时器
                //使用定时任务发送心跳消息
                ScheduledFuture<?> heartBeat = ctx.executor().scheduleAtFixedRate(new NetLiveHandler.HeartBeatTask(ctx,connectType), 0, 15000, TimeUnit.MILLISECONDS);
                String taskKey = ctx.channel().remoteAddress().toString();
                sessionRepository.getTaskMap().put(taskKey,heartBeat);
//                ctx.executor().shutdownGracefully();
                logger.info(" \n       start schedule task ------》》》》》》》》" + heartBeat.toString());
            }
            // 收到心跳消息，Pong
            else if (Action.HEART_BEAT.getName().equalsIgnoreCase(action)) {
//                TcpResponse heatBeat = new TcpResponse(Action.HEART_BEAT.getName());
//                heatBeat.setMessage("Pong");
//                String heatBeatMsg = JsonUtil.toJson(heatBeat);
//                logger.trace("Response heart beat messsage to client ---> " + heatBeatMsg);
//                ctx.channel().writeAndFlush(heatBeatMsg + Constants.delimiter);
            } else {
                // 业务逻辑处理
                liveService.process(ctx, req, action,connectType);
            }
        } catch (NtClientException e) {
            logger.error(e.getMessage(), e);
            
            /**
             * resive by C.H on 2018/10/29 11:23
             * 修改：
             */
            ChannelUtil.get(connectType).writeAndFlush(ctx.channel(),e.toJson() + Constants.delimiter);
//          ctx.channel().writeAndFlush(e.toJson() + Constants.delimiter);
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
            TcpResponse response = new TcpResponse(Action.CREATE_SESSION.getName());
            response.setMessage("Server Error");
            response.setStatus(500);

            /**
             * resive by C.H on 2018/10/29 11:24
             * 修改：
             */
            ChannelUtil.get(connectType).writeAndFlush(ctx.channel(),JsonUtil.toJson(response) + Constants.delimiter);
//            ctx.channel().writeAndFlush(JsonUtil.toJson(response) + Constants.delimiter);
        }

    }
    private Integer getIntegerForRoomId(String action, Object objRoomId) throws NtClientException {
        Integer roomId = null;
        try {
            roomId = Integer.parseInt(objRoomId.toString());
        } catch (NumberFormatException e) {
            logger.error("RoomId must be number, roomId=" + objRoomId.toString(), e);
            throw new NtClientException("RoomId must be number, roomId=" + objRoomId.toString(), action);
        }
        return roomId;
    }
    private void createSession(ChannelHandlerContext ctx, TcpRequest req, String action, Integer roomId, Integer connectType) throws NtClientException {
        int role = req.getHead().getNtClientRole();
        String token = req.getHead().getNtToken();
        Integer clientOs = req.getHead().getNtClientOSType();
        Channel channel = ctx.channel();

        // 如果第一个用户创建session,那么需要创建房间Session,然后将用户Session放到房间里
        Room roomSession = sessionRepository.get(roomId);
        if (roomSession == null) {
            roomSession = new Room(roomId);
            logger.info("Created roomSession，roomId={}, endpoint:{}", roomId, channel.remoteAddress().toString());
            ClassRoomEntryLog roomLog = new ClassRoomEntryLog();
            roomLog.setRoom_id(roomId);
            roomLog.setEvent_type("createTcpRoomSession");
            roomLog.setEvent_date(new Date());
            roomLog.setUser_role(role);
            roomLog.setClient_os_type(clientOs);
            roomLog.setDetail("Created TCP Room session with " + role + "endPoint:" + channel.remoteAddress().toString() + " for os:" + clientOs);
            liveService.roomLog(roomLog);
        }

        if (role == Constants.UserRole.STUDENT) {
            Member member = liveService.getMemberByToken(token);
            if (member == null) {
                throw new NtClientException("Invalid token:" + token + " for member", action);
            }

            User user = new User(member, clientOs);
            // 记录session创建日志
            ClassRoomEntryLog roomLog = new ClassRoomEntryLog();
            roomLog.setRoom_id(roomId);
            roomLog.setEvent_type("createTcpSession");
            roomLog.setEvent_date(new Date());
            roomLog.setUser_id(user.getId());
            roomLog.setUser_role(role);
            roomLog.setClient_os_type(clientOs);
            if (roomSession.getStudent() == null) {
                logger.info("Created TCP room session for student:(id:{}-name:{}-token:{})，roomId={}, endpoint:{}", user.getId(), user.getName(), user.getToken(), roomId, channel.remoteAddress().toString());
                roomLog.setDetail("Created TCP room session(Student),endPoint:" + channel.remoteAddress().toString() + " for OS:" + clientOs);
            } else {
                logger.info("Created other TCP room session for student:(id:{}-name:{}-token:{})，roomId={}, endpoint:{}", user.getId(), user.getName(), user.getToken(), roomId, channel.remoteAddress().toString());
                roomLog.setDetail("Created other TCP room session(Student),endPoint:" + channel.remoteAddress().toString() + " for OS:" + clientOs);
                logger.info("old student endpoint [{}]",roomSession.getStudent().getChannel().remoteAddress().toString());
                logger.info("new student endpoint [{}]",channel.remoteAddress().toString());
                // 发消息，通知上一个客户端退出系统
                TcpResponse response = new TcpResponse(Action.LOGOUT.getName());
                response.setMessage("您已在新的设备登录，该设备退出登录。");
                final Session oldSession = roomSession.getStudent();

                /**
                 * resive by C.H on 2018/10/29 10:47
                 * 修改：
                 */
                ChannelUtil.get(oldSession.getConnectType()).writeAndFlush(oldSession.getChannel(),JsonUtil.toJson(response) + Constants.delimiter).addListener(new ChannelFutureListener() {
//                oldSession.getChannel().writeAndFlush(JsonUtil.toJson(response) + Constants.delimiter).addListener(new ChannelFutureListener() {
                    @Override
                    public void operationComplete(ChannelFuture future) throws Exception {
                        if (future.isSuccess()) {
                            logger.info("通知退出成功 to: " + oldSession.getChannel().remoteAddress() + " User:" + JsonUtil.toJson(oldSession.getUser()));
                        } else {
                            logger.error("通知退出失败 to: " + oldSession.getChannel().remoteAddress() + " 原因为:", future.cause() + " User:" + JsonUtil.toJson(oldSession.getUser()));
                        }
                    }
                });
                oldSession.getChannel().close();

            }


            Session session = new Session(channel, user,connectType);
            roomSession.setStudent(session);
            liveService.roomLog(roomLog);

        } else if (role == Constants.UserRole.TEACHER) {
            Teacher teacher = liveService.getTeacherByToken(token);
            if (teacher == null) {
                throw new NtClientException("Invalid token:" + token + " for teacher", action);
            }
            User user = new User(teacher, clientOs);
            // 记录session创建日志
            ClassRoomEntryLog roomLog = new ClassRoomEntryLog();
            roomLog.setRoom_id(roomId);
            roomLog.setEvent_type("createTcpSession");
            roomLog.setEvent_date(new Date());
            roomLog.setUser_id(user.getId());
            roomLog.setUser_role(role);
            roomLog.setClient_os_type(clientOs);
            if (roomSession.getTeacher() == null) {
                logger.info("Created TCP Teacher session for teacher:(id:{}-name:{}-token:{})，roomId={}, endpoint:{}", user.getId(), user.getName(), user.getToken(), roomId, channel.remoteAddress().toString());
                roomLog.setDetail("Created TCP Teacher session(Teacher),endPoint:" + channel.remoteAddress().toString() + " for os:" + clientOs);
            } else {
                logger.info("Created other TCP Teacher session for teacher:(id:{},name:{},token:{})，roomId={}, endpoint:{}", user.getId(), user.getName(), user.getToken(), roomId, channel.remoteAddress().toString());
                logger.info("old teacher endpoint [{}]",roomSession.getTeacher().getChannel().remoteAddress().toString());
                logger.info("new teacher endpoint [{}]",channel.remoteAddress().toString());
                roomLog.setDetail("Created other TCP Teacher session(Teacher),endPoint:" + channel.remoteAddress().toString() + " for os:" + clientOs);
                // 发消息，通知上一个客户端退出系统
                TcpResponse response = new TcpResponse(Action.LOGOUT.getName());
                response.setMessage("您已在新的设备登录，该设备退出登录。");
                final Session oldSession = roomSession.getTeacher();
                /**
                 * resive by C.H on 2018/10/29 10:52
                 * 修改：
                 */
                ChannelUtil.get(oldSession.getConnectType()).writeAndFlush(oldSession.getChannel(),JsonUtil.toJson(response) + Constants.delimiter).addListener(new ChannelFutureListener() {
//                oldSession.getChannel().writeAndFlush(JsonUtil.toJson(response) + Constants.delimiter).addListener(new ChannelFutureListener() {
                    @Override
                    public void operationComplete(ChannelFuture future) throws Exception {
                        if (future.isSuccess()) {
                            logger.info("通知退出成功 to: " + oldSession.getChannel().remoteAddress() + " User:" + JsonUtil.toJson(oldSession.getUser()));
                        } else {
                            logger.error("通知退出失败 to: " + oldSession.getChannel().remoteAddress() + " 原因为:", future.cause() + " User:" + JsonUtil.toJson(oldSession.getUser()));
                        }
                    }
                });
                oldSession.getChannel().close();
            }


            Session session = new Session(channel, user,connectType);
            roomSession.setTeacher(session);
            liveService.roomLog(roomLog);
        } else {
            throw new NtClientException("Invalid ntClientRole", action);
        }

        // 创建教室级别的session
        sessionRepository.put(roomId, roomSession);

        logger.debug("Current room size is {}", this.sessionRepository.size());
    }

    private void validateTcpHead(TcpRequestHead head) throws NtClientException {
        if (head == null) {
            throw new NtClientException("Invalid data format:head is required");
        }
        String apiVersion = head.getNtApiVersion();
        if (StringUtils.isBlank(apiVersion)) {
            throw new NtClientException("Invalid data format:ntApiVersion is required");
        }

        Long timestamp = head.getNtTimestamp();
        if (timestamp == null) {
            throw new NtClientException("Invalid data format:ntTimestamp is required");
        }

        String language = head.getNtClientLanguage();
        if (StringUtils.isBlank(language)) {
            throw new NtClientException("Invalid data format:ntClientLanguage is required");
        }

        String signature = head.getNtSignature();
        if (StringUtils.isBlank(signature)) {
            throw new NtClientException("Invalid data format:ntSignature is required");
        }

        String token = head.getNtToken();
        if (StringUtils.isBlank(token)) {
            throw new NtClientException("Invalid data format:token is required");
        }

        Integer osType = head.getNtClientOSType();
        if (osType == null) {
            throw new NtClientException("Invalid data format:ntClientOSType is required");
        }

        Integer clientType = head.getNtClientType();
        if (clientType == null) {
            throw new NtClientException("Invalid data format:ntClientType is required");
        }

        Integer role = head.getNtClientRole();
        if (role == null) {
            throw new NtClientException("Invalid data format:ntClientRole is required");
        }


    }


    //Ping消息任务类
    private class HeartBeatTask implements Runnable {
        private final ChannelHandlerContext ctx;
        private volatile Session session;
        private int connectType;

        public HeartBeatTask(final ChannelHandlerContext ctx,int connectType) {
            this.ctx = ctx;
            this.session = sessionRepository.getSessionByKey(ctx.channel().remoteAddress().toString());
            this.connectType=connectType;
        }

        @Override
        public void run() {

            TcpResponse heatBeat = new TcpResponse(Action.HEART_BEAT.getName());
            //发送心跳并添加一个侦听器，如果发送操作失败将关闭连接
            String heatBeatMsg = JsonUtil.toJson(heatBeat);
            heatBeat.setMessage("Ping");
            logger.info("##### endpoint[{}]status -->> isActive:{},isOpen:{},isRegistered:{},isWritable:{},",ctx.channel().remoteAddress().toString(),ctx.channel().isActive(), ctx.channel().isOpen(),ctx.channel().isRegistered(),ctx.channel().isWritable());
            /**
             * resive by C.H on 2018/10/29 11:26
             * 修改：
             */
            ChannelUtil.get(connectType).writeAndFlush(ctx.channel(), heatBeatMsg + Constants.delimiter).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
//          ctx.channel().writeAndFlush(heatBeatMsg + Constants.delimiter).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);

            String clientEndPoint = ctx.channel().remoteAddress().toString();
            if (session == null) {
                session = sessionRepository.getSessionByKey(clientEndPoint);
            }
            if (session != null) {
                User user = session.getUser();
                heatBeat.setMessage("Ping to " + user.getUserRole() + ":" + user.getName());
                logger.info("\n    =========>>>:: Send heartbeat to client[" + clientEndPoint + "] --> " + JsonUtil.toJson(user));
            } else {
                logger.info("\n    =========>>>:: Send heartbeat to client[" + clientEndPoint + "] 【None session】");
            }
        }

    }

    public void setChannelRepository(SessionRepository sessionRepository) {
        this.sessionRepository = sessionRepository;
    }

}
